package pt.compflow.activity.task;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import pt.compflow.Context;
import pt.compflow.activity.ActivityControllerListener;
import pt.compflow.flowontology.model.Assignment;
import pt.compflow.flowontology.model.FLOWO;
import pt.compflow.flowontology.model.FlowOntologyModel;
import pt.compflow.flowontology.model.Interface;
import pt.compflow.flowontology.model.Job;
import pt.compflow.flowontology.model.Task;
import pt.compflow.flowontology.model.TaskInterface;
import pt.compflow.flowontology.model.Worker;
import pt.compflow.flowontology.model.Workflow;
import pt.compflow.graph.Tree;
import pt.compflow.interfaces.IOInterface;
import pt.compflow.storage.TripleStoreConnection;
import pt.compflow.tools.OntModelHelper;
import pt.ipp.isep.gecad.o2c.model.impl.DateContainer;
import pt.ipp.isep.gecad.o2c.model.impl.DateContainer.DateContainerType;

import com.hp.hpl.jena.rdf.model.Resource;
import com.hp.hpl.jena.rdf.model.Statement;
import com.hp.hpl.jena.rdf.model.StmtIterator;
import com.hp.hpl.jena.util.iterator.ExtendedIterator;
import com.hp.hpl.jena.vocabulary.RDF;

/**
 * Handles all Tasks and Assignments
 *
 */
public class TaskControllerImpl implements TaskController {
		
	private static final Logger LOG = LoggerFactory.getLogger(TaskControllerImpl.class);
	
	// listeners of the TaskController; these will be notified when events regarding the Task occur
	protected List<ActivityControllerListener> listeners;
	protected Random random = new Random();

	// context object
	protected Context context;
	
	public TaskControllerImpl(Context context) {
		this.context = context;
	}
	
	protected void restartPending() {
		TripleStoreConnection store = context.getStoreConnection();
		store.openStore();
		
		FlowOntologyModel engineModel = store.connectModel();
		
		Iterator<Job> it = engineModel.listJobs();
		
		while(it.hasNext()) {
			Job job = it.next();
			String jobUri = job.getURI();
		
			FlowOntologyModel jobModel = store.connectNamedModel(jobUri);
			
			Iterator<Task> tIt = jobModel.listTasks();
			while(tIt.hasNext()) {
			
				Task t = tIt.next();
				Workflow wf = t.getActivityOfWorkflow();
				
				if(t.getPropertyResourceValue(FLOWO.hasState).getURI().compareTo(FLOWO.inProgress.getURI()) == 0 && 
						wf.getPropertyResourceValue(FLOWO.hasState).getURI().compareTo(FLOWO.inProgress.getURI()) == 0) {
										
					// get the ontology model (workflow-definition) of this workflow
					OntModelHelper ontModelHelper = context.getOntModelHelperManager().get(store, jobUri, wf.getRootWorkflow().getLocalName());
					
					if(!t.listHasAssignmentAssignments().hasNext())
						context.getInputFinder().setInputAndContext(context, ontModelHelper, engineModel, jobModel, job, wf, t);
										
					Map<String, Integer> assignMap = new HashMap<String, Integer>();
					Iterator<Assignment> aIt = t.listHasAssignmentAssignments();
					while(aIt.hasNext()) {
						Assignment a = aIt.next();
						String inputUri = a.getHasInput() == null ? "default" : a.getHasInput().getURI();
						if(!assignMap.containsKey(inputUri)) assignMap.put(inputUri, 1);
						else assignMap.put(inputUri, assignMap.get(inputUri) + 1);
					}
							
					Resource taskClass = ontModelHelper.getDirectLeafType(t, FLOWO.Task.getURI());
					Tree assignmentTree = ontModelHelper.getApplicableEdgeDestinationsClosure(FLOWO.hasAssignment.getURI(), taskClass.getURI(), FLOWO.Task.getURI());
									
					// creating minimum required assignments for each Unit in the Task (no workers assigned yet; just units)
					// the META_INPUT_PROP triple is set by the ActivityInstantiator during the instantiation of the task
					// it contains the property that relates the task to its input (a subPropertyOf hasInput)
			//		Property inputProperty = t.getPropertyResourceValue(jobModel.getProperty(META_INPUT_PROP)).as(Property.class);
					List<Statement> unitList = t.listProperties(FLOWO.hasInput).toList();	// toList to avoid concurrent exceptions...
					// numberOfCreatedAssignments = assignmentsPerUnit * amountOfInputUnits_(amountOfInputInstances)
					int au = t.getHasAssignmentsPerUnit();
					for(Statement stmt : unitList) {
						Resource unit = stmt.getObject().asResource();
						int cau = assignMap.containsKey(unit.getURI()) ? assignMap.get(unit.getURI()) : 0;
						
						for(int i=cau;i<au;i++) {
							Assignment assignment = jobModel.createAssignmentWithURI(FLOWO.getURI() + context.getInstantiator().getUniqueAssignmentId(jobModel, null, null, null));
							assignment.setHasCreationDate(new DateContainer(DateContainerType.DATETIME));
							assignment.addProperty(FLOWO.hasPriority, t.getPropertyResourceValue(FLOWO.hasPriority) == null ? wf.getPropertyResourceValue(FLOWO.hasPriority) : t.getPropertyResourceValue(FLOWO.hasPriority));
							assignment.removeAll(FLOWO.hasState);
							assignment.addProperty(FLOWO.hasState, FLOWO.notStarted);
							assignment.setHasTimeout(t.getHasTimeout());
							
							if(assignmentTree != null && !assignmentTree.isEmpty()) {
								for(Tree assignmentSubTree : assignmentTree) 
									assignment.addProperty(RDF.type, jobModel.getResource(assignmentSubTree.getNode()));
							}
							
							context.getInputFinder().setInputAndContext(context, ontModelHelper, engineModel, jobModel, job, wf, t, assignment, unit);
							assignment.addAssignmentOfTask(t);	// set the task of the assignment
							t.addHasAssignmentAssignment(assignment);	// set the assignment to the task
						}
					}	
					
					// request (broadcast) task through all available interfaces
					StmtIterator taskInterfaceIt = t.listProperties(FLOWO.deliveredThrough);
					while(taskInterfaceIt.hasNext()) {
						TaskInterface taskInterface = engineModel.getTaskInterfaceWithURI(taskInterfaceIt.next().getObject().asResource().getURI());		
						Resource taskInterfaceClass = ontModelHelper.getDirectLeafType(taskInterface, FLOWO.TaskInterface.getURI());
						IOInterface taskInterfaceHandler = context.getTaskInterfaceManager().get(taskInterface, taskInterfaceClass, ontModelHelper.getOntModel(), context, store);
						
						// request the task to the taskInterface (async command)
						Runnable requestTaskCommand = new TaskControllerRequestTaskCommand(context, this, taskInterfaceHandler, jobUri, t.getURI());
						context.getExecutorService().execute(requestTaskCommand);
					}
							
					LOG.debug("Re-Started Task " + t.getURI());
							
					if(unitList.size() <= 0) {
						LOG.debug("No units were found for the task " + t.getURI());
						notifyTaskFinished(jobUri, t.getURI());
					}
				}
			}
		}
		
		store.closeStore();
		
	}
	
	@Override
	public void start() {
		// register the controller as a listener of task interfaces
		context.getTaskInterfaceManager().addInterfaceListener(this);
		
		restartPending();
	}

	@Override
	public void stop() {
		// unregister the controller as a listener of task interfaces
		context.getTaskInterfaceManager().removeInterfaceListener(this);
	}

	/**
	 * Notifies all listeners that a task has finished
	 * @param taskUri of the finished task
	 */
	protected void notifyTaskFinished(String jobUri, String taskUri) {		
		if(listeners == null) return;
		for(ActivityControllerListener l : listeners)
			l.onActivityFinished(jobUri, taskUri, null);
	}
	
	/**
	 * Notifies all listeners that a task has started
	 * @param taskUri of the started task
	 */
	protected void notifyTaskStarted(String jobUri, String taskUri) {		
		if(listeners == null) return;
		for(ActivityControllerListener l : listeners)
			l.onActivityStarted(jobUri, taskUri);
	}
	
	/**
	 * Notifies all listeners that a task has errors
	 * @param taskUri of the task with errors
	 */
	protected void notifyTaskError(String jobUri, String taskUri) {		
		if(listeners == null) return;
		for(ActivityControllerListener l : listeners)
			l.onActivityError(jobUri, taskUri);
	}
	
	/**
	 * Notifies all listeners that a task has been cancelled
	 * @param taskUri of the cancelled task
	 */
	protected void notifyTaskCancelled(String jobUri, String taskUri) {		
		if(listeners == null) return;
		for(ActivityControllerListener l : listeners)
			l.onActivityCancelled(jobUri, taskUri);
	}
	
	/**
	 * For the given Task, request the first assignment with state "notStarted" that is found
	 * @param jobModel - the instance data FlowOntologyModel model
	 * @param wf - the Workflow
	 * @param t - the Task
	 * @param worker - the Worker to be requested
	 * @param interfaze - the Interface with the Worker
	 */
	protected synchronized boolean requestNotStartedInProgressAssignment(TripleStoreConnection store, OntModelHelper ontModelHelper, FlowOntologyModel jobModel, String jobUri, Workflow wf, Task t, Worker worker, Interface interfaze) {
		
		LOG.debug("Requesting Assignments for Task " + t.getURI() + " and Worker " + worker.getURI() + " through " + interfaze.getURI());
				
		Set<String> doneUnitUris = new HashSet<String>();
		StmtIterator sIt = jobModel.listStatements(null, FLOWO.performedBy, worker);
		while(sIt.hasNext()) {
			Statement stm = sIt.next();
			if(stm.getSubject().canAs(Assignment.class) && t.hasProperty(FLOWO.hasAssignment, stm.getSubject())) {
				doneUnitUris.add(stm.getSubject().as(Assignment.class).getHasInput().getURI());
			}
		}
		
		// for the Task t, get the first Assignment with state notStarted
		boolean redundant = false;
		Assignment assignment = null;
		ExtendedIterator<Assignment> assignmentIt = t.listHasAssignmentAssignments();
		List<Assignment> inProgressList = new ArrayList<Assignment>();
		while(assignmentIt.hasNext() && assignment == null) {
			Assignment itAssignment = assignmentIt.next();
			if(itAssignment.getPropertyResourceValue(FLOWO.hasState) != null) {
				
				if(itAssignment.getPropertyResourceValue(FLOWO.hasState).getURI().compareToIgnoreCase(FLOWO.notStarted.getURI()) == 0)
					if(!doneUnitUris.contains(itAssignment.getHasInput().getURI())) assignment = itAssignment; else redundant = true;
				else if(itAssignment.getPropertyResourceValue(FLOWO.hasState).getURI().compareToIgnoreCase(FLOWO.inProgress.getURI()) == 0)
					if(!doneUnitUris.contains(itAssignment.getHasInput().getURI())) inProgressList.add(itAssignment); else redundant = true;
				
			}
		}
		
		// randomly request an assignment that has been previously delivered
		// but not yet solved
		if(assignment == null && inProgressList.size() > 0) {
			assignment = inProgressList.get(random.nextInt(inProgressList.size()));
		}
		
		// if a pending (notStarted) assignment is found...
		if(assignment != null) {
			// get the explicit direct type of the task interface
			Resource taskInterfaceClass = ontModelHelper.getDirectLeafType(interfaze, FLOWO.TaskInterface.getURI());
					
			// set assignment state inProgress
			assignment.removeAll(FLOWO.hasState);
			assignment.addProperty(FLOWO.hasState, FLOWO.inProgress);
			
			// get the interface handler instance mapped to the interface ontology class
			IOInterface taskInterface = context.getTaskInterfaceManager().get(interfaze, taskInterfaceClass, ontModelHelper.getOntModel(), context, store);
						
			// request the assignment to the worker through the interface (async command)
			Runnable requestAssignmentCommand = new TaskControllerRequestAssignmentCommand(context, this, taskInterface, jobUri, t.getURI(), assignment.getURI(), worker.getURI());
			context.getExecutorService().execute(requestAssignmentCommand);	
			
			return true;
		} 
		
		// do not finish task but also do not request assignments to this worker
		if(redundant) return false;
		
		LOG.debug("No more pending assignments for Task " + t.getURI());
		
		LOG.debug("Solved Task " + t.getURI() + " by " + worker.getURI() + " through " + interfaze.getURI());
		notifyTaskFinished(jobUri, t.getURI());
		
		return false;
	}

	@Override
	public void startTask(String jobUri, String taskUri) {
		
		TripleStoreConnection store = context.getStoreConnection();
		store.openStore();
		
		FlowOntologyModel engineModel = store.connectModel();
		FlowOntologyModel jobModel = store.connectNamedModel(jobUri);
		
		Task t = jobModel.getTaskWithURI(taskUri);
		Workflow wf = t.getActivityOfWorkflow();
				
		// the task must be in state "notStarted" for it to be started...
		if(!t.getPropertyResourceValue(FLOWO.hasState).getURI().equals(FLOWO.notStarted.getURI())) {
			LOG.warn("Task state is not valid for a Start Command; Nothing done. State = " + t.getPropertyResourceValue(FLOWO.hasState).getURI());
			store.closeStore();
			return;
		}
		
		Job job = engineModel.getJobWithURI(jobUri);
		
		// get the ontology model (workflow-definition) of this workflow
		OntModelHelper ontModelHelper = context.getOntModelHelperManager().get(store, jobUri, wf.getRootWorkflow().getLocalName());
		
		context.getInputFinder().setInputAndContext(context, ontModelHelper, engineModel, jobModel, job, wf, t);
						
		// setup task state and start date
		t.setHasStartDate(new DateContainer(DateContainerType.DATETIME));
		t.removeAll(FLOWO.hasState);
		t.addProperty(FLOWO.hasState, FLOWO.inProgress);
		
		Resource taskClass = ontModelHelper.getDirectLeafType(t, FLOWO.Task.getURI());
		Tree assignmentTree = ontModelHelper.getApplicableEdgeDestinationsClosure(FLOWO.hasAssignment.getURI(), taskClass.getURI(), FLOWO.Task.getURI());
						
		// creating minimum required assignments for each Unit in the Task (no workers assigned yet; just units)
		// the META_INPUT_PROP triple is set by the ActivityInstantiator during the instantiation of the task
		// it contains the property that relates the task to its input (a subPropertyOf hasInput)
//		Property inputProperty = t.getPropertyResourceValue(jobModel.getProperty(META_INPUT_PROP)).as(Property.class);
		List<Statement> unitList = t.listProperties(FLOWO.hasInput).toList();	// toList to avoid concurrent exceptions...
		// numberOfCreatedAssignments = assignmentsPerUnit * amountOfInputUnits_(amountOfInputInstances)
		for(int i=0;i<t.getHasAssignmentsPerUnit();i++) {
			for(Statement stmt : unitList) {
				Resource unit = stmt.getObject().asResource();
				Assignment assignment = jobModel.createAssignmentWithURI(FLOWO.getURI() + context.getInstantiator().getUniqueAssignmentId(jobModel, null, null, null));
				assignment.setHasCreationDate(new DateContainer(DateContainerType.DATETIME));
				assignment.addProperty(FLOWO.hasPriority, t.getPropertyResourceValue(FLOWO.hasPriority) == null ? wf.getPropertyResourceValue(FLOWO.hasPriority) : t.getPropertyResourceValue(FLOWO.hasPriority));
				assignment.removeAll(FLOWO.hasState);
				assignment.addProperty(FLOWO.hasState, FLOWO.notStarted);
				assignment.setHasTimeout(t.getHasTimeout());
				
				if(assignmentTree != null && !assignmentTree.isEmpty()) {
					for(Tree assignmentSubTree : assignmentTree) 
						assignment.addProperty(RDF.type, jobModel.getResource(assignmentSubTree.getNode()));
				}
				
				context.getInputFinder().setInputAndContext(context, ontModelHelper, engineModel, jobModel, job, wf, t, assignment, unit);
				assignment.addAssignmentOfTask(t);	// set the task of the assignment
				t.addHasAssignmentAssignment(assignment);	// set the assignment to the task
			}
		}	
		
		// request (broadcast) task through all available interfaces
		StmtIterator taskInterfaceIt = t.listProperties(FLOWO.deliveredThrough);
		while(taskInterfaceIt.hasNext()) {
			TaskInterface taskInterface = engineModel.getTaskInterfaceWithURI(taskInterfaceIt.next().getObject().asResource().getURI());		
			Resource taskInterfaceClass = ontModelHelper.getDirectLeafType(taskInterface, FLOWO.TaskInterface.getURI());
			IOInterface taskInterfaceHandler = context.getTaskInterfaceManager().get(taskInterface, taskInterfaceClass, ontModelHelper.getOntModel(), context, store);
			
			// request the task to the taskInterface (async command)
			Runnable requestTaskCommand = new TaskControllerRequestTaskCommand(context, this, taskInterfaceHandler, jobUri, taskUri);
			context.getExecutorService().execute(requestTaskCommand);
		}
				
		LOG.debug("Started Task " + taskUri);
		
		// notify listeners that the task has started
		notifyTaskStarted(jobUri, taskUri);
		
		if(unitList.size() <= 0) {
			LOG.debug("No units were found for the task " + t.getURI());
			notifyTaskFinished(jobUri, t.getURI());
		}
		
		store.closeStore();
	}
	
	@Override
	public void cancelTask(String jobUri, String taskUri, boolean isTimeout) {
		
		TripleStoreConnection store = context.getStoreConnection();
		store.openStore();
		
		LOG.debug("Cancelling Task " + taskUri + (isTimeout ? " (TIMEDOUT)" : ""));
		
		FlowOntologyModel engineModel = store.connectModel();
		FlowOntologyModel jobModel = store.connectNamedModel(jobUri);
		
		Task t = jobModel.getTaskWithURI(taskUri);
		Workflow wf = t.getActivityOfWorkflow();
				
		// get the ontology model (workflow-definition) of this workflow
		OntModelHelper ontModelHelper = context.getOntModelHelperManager().get(store, jobUri, wf.getRootWorkflow().getLocalName());
		
		// cancel (broadcast) task through all available interfaces
		StmtIterator taskInterfaceIt = t.listProperties(FLOWO.deliveredThrough);
		while(taskInterfaceIt.hasNext()) {
			TaskInterface taskInterface = engineModel.getTaskInterfaceWithURI(taskInterfaceIt.next().getObject().asResource().getURI());			
			Resource taskInterfaceClass = ontModelHelper.getDirectLeafType(taskInterface, FLOWO.TaskInterface.getURI());
			IOInterface taskInterfaceHandler = context.getTaskInterfaceManager().get(taskInterface, taskInterfaceClass, ontModelHelper.getOntModel(), context, store);
			
			// cancel the task with the taskInterface (async command) this command does not cause onCommandError calls
			// errors in this command are just logged
			Runnable cancelTaskCommand = new TaskControllerCancelTaskCommand(context, this, taskInterfaceHandler, jobUri, taskUri);
			context.getExecutorService().execute(cancelTaskCommand);
		}
		
		// set task and pending assignments with cancelled state
		t.removeAll(FLOWO.hasState);
		t.addProperty(FLOWO.hasState, isTimeout ? FLOWO.timedout : FLOWO.canceled);
		
		ExtendedIterator<Assignment> aIt = t.listHasAssignmentAssignments();
		while(aIt.hasNext()) {
			Assignment a = aIt.next();
			if(a.getPropertyResourceValue(FLOWO.hasState).getURI().compareTo(FLOWO.notStarted.getURI()) == 0 || a.getPropertyResourceValue(FLOWO.hasState).getURI().compareTo(FLOWO.inProgress.getURI()) == 0) {
				a.removeAll(FLOWO.hasState);
				a.addProperty(FLOWO.hasState, isTimeout ? FLOWO.timedout : FLOWO.canceled);
			}
		}
				
		if(!isTimeout) {
			// notify workflow controller of cancellation in task
			notifyTaskCancelled(jobUri, taskUri);	
		}
		
		store.closeStore();
	}
		
	@Override
	public boolean onTaskAccepted(String jobUri, String taskUri, String workerUri, String interfaceUri) {
		TripleStoreConnection store = context.getStoreConnection();
		store.openStore();
		
		FlowOntologyModel engineModel = store.connectModel();
		FlowOntologyModel jobModel = store.connectNamedModel(jobUri);
		
		Task t = jobModel.getTaskWithURI(taskUri);
		Workflow wf = t.getActivityOfWorkflow();
		
		Worker worker = engineModel.getWorkerWithURI(workerUri);
		Interface interfaze = engineModel.getInterfaceWithURI(interfaceUri);
		
		boolean ok = false;
		if(t.getPropertyResourceValue(FLOWO.hasState).getURI().compareTo(FLOWO.inProgress.getURI()) != 0) {
			LOG.warn("Task " + taskUri + " state is invalid (" + t.getPropertyResourceValue(FLOWO.hasState).getLocalName() + ")! Not in progress? Ignoring task acceptance...");
		} else {
			t.addAcceptedByWorker(worker);
			
			LOG.debug("Accepted Task " + taskUri + " by " + workerUri + " through " + interfaceUri);
					
			// request the worker to solve a notStarted assignment
			// deliver the assignment through the interface
			OntModelHelper ontModelHelper = context.getOntModelHelperManager().get(store, jobUri, wf.getRootWorkflow().getLocalName());
		
			ok = requestNotStartedInProgressAssignment(store, ontModelHelper, jobModel, jobUri, wf, t, worker, interfaze);
		}
		
		store.closeStore();
		
		return ok;
	}
	
	@Override
	public boolean onAssignmentSolved(String jobUri, String taskUri, String assignmentUri, String workerUri, String interfaceUri) {
		TripleStoreConnection store = context.getStoreConnection();
		store.openStore();
		
		FlowOntologyModel engineModel = store.connectModel();
		FlowOntologyModel jobModel = store.connectNamedModel(jobUri);
		
		Task t = jobModel.getTaskWithURI(taskUri);
		Workflow wf = t.getActivityOfWorkflow();
		Worker worker = engineModel.getWorkerWithURI(workerUri);
		Interface interfaze = engineModel.getInterfaceWithURI(interfaceUri);
		
		if(!t.hasProperty(FLOWO.acceptedBy, worker)) {
			LOG.warn("Given assignment " + assignmentUri + " response has been ignored! A worker cannot solve assignments before accepting the task (" + taskUri + ")!");
			store.closeStore();
			return false;
		}
		
		Assignment assignment = jobModel.getAssignmentWithURI(assignmentUri);
		if(assignment.getPropertyResourceValue(FLOWO.hasState).getURI().compareTo(FLOWO.inProgress.getURI()) != 0) {
			LOG.warn("Assignment " + assignmentUri + " state is invalid (" + assignment.getPropertyResourceValue(FLOWO.hasState).getLocalName() + ")! Already solved? Ignoring given solution...");
		} else {
			// set the assignment properties
			assignment.setPerformedByWorker(worker);
			assignment.setDeliveredThroughInterface(interfaze);
			assignment.removeAll(FLOWO.hasState);
			assignment.addProperty(FLOWO.hasState, FLOWO.finished);
			assignment.setHasCompletionDate(new DateContainer(DateContainerType.DATETIME));
		}
		
		// cancel the assignment for all other interfaces since it has already been solved
		OntModelHelper ontModelHelper = context.getOntModelHelperManager().get(store, jobUri, wf.getRootWorkflow().getLocalName());
		
		StmtIterator taskInterfaceIt = t.listProperties(FLOWO.deliveredThrough);
		while(taskInterfaceIt.hasNext()) {
			TaskInterface i = engineModel.getTaskInterfaceWithURI(taskInterfaceIt.next().getObject().asResource().getURI());
			if(i.getURI().compareTo(interfaceUri) != 0) {
				Resource taskInterfaceClass = ontModelHelper.getDirectLeafType(i, FLOWO.TaskInterface.getURI());
				IOInterface taskInterfaceHandler = context.getTaskInterfaceManager().get(i, taskInterfaceClass, ontModelHelper.getOntModel(), context, store);
				
				TaskControllerCancelAssignmentCommand cancelAssignmentCommand = new TaskControllerCancelAssignmentCommand(context, this, taskInterfaceHandler, jobUri, taskUri, assignmentUri);
				context.getExecutorService().execute(cancelAssignmentCommand);
			}
		}
	
		// request the next pending assignment to this worker if available
		boolean ok = requestNotStartedInProgressAssignment(store, ontModelHelper, jobModel, jobUri, wf, t, worker, interfaze);
		
		store.closeStore();
		
		return ok;
	}
		
	@Override
	public void onCommandError(String jobUri, String taskUri, String interfaceUri) {
		
		TripleStoreConnection store = context.getStoreConnection();
		store.openStore();
		
		FlowOntologyModel jobModel = store.connectNamedModel(jobUri);
		FlowOntologyModel engineModel = store.connectModel();
		
		Task t = jobModel.getTaskWithURI(taskUri);
		Workflow wf = t.getActivityOfWorkflow();
				
		// get the ontology model (workflow-definition) of this workflow
		OntModelHelper ontModelHelper = context.getOntModelHelperManager().get(store, jobUri, wf.getRootWorkflow().getLocalName());
		
		// cancel (broadcast) task through all available interfaces
		StmtIterator taskInterfaceIt = t.listProperties(FLOWO.deliveredThrough);
		while(taskInterfaceIt.hasNext()) {
			TaskInterface taskInterface = engineModel.getTaskInterfaceWithURI(taskInterfaceIt.next().getObject().asResource().getURI());
			Resource taskInterfaceClass = ontModelHelper.getDirectLeafType(taskInterface, FLOWO.TaskInterface.getURI());
			IOInterface taskInterfaceHandler = context.getTaskInterfaceManager().get(taskInterface, taskInterfaceClass, ontModelHelper.getOntModel(), context, store);
			
			// cancel the task with the taskInterface (async command) this command does not cause onCommandError calls
			// errors in this command are just logged
			Runnable cancelTaskCommand = new TaskControllerCancelTaskCommand(context, this, taskInterfaceHandler, jobUri, taskUri);
			context.getExecutorService().execute(cancelTaskCommand);
		}
		
		// set task and pending assignments with withError state
		t.removeAll(FLOWO.hasState);
		t.addProperty(FLOWO.hasState, FLOWO.withError);
		ExtendedIterator<Assignment> aIt = t.listHasAssignmentAssignments();
		while(aIt.hasNext()) {
			Assignment a = aIt.next();
			if(a.getPropertyResourceValue(FLOWO.hasState).getURI().compareTo(FLOWO.notStarted.getURI()) == 0 || a.getPropertyResourceValue(FLOWO.hasState).getURI().compareTo(FLOWO.inProgress.getURI()) == 0) {
				a.removeAll(FLOWO.hasState);
				a.addProperty(FLOWO.hasState, FLOWO.withError);
			}
		}
		
		LOG.error("Error interacting through the TaskInterface " + interfaceUri + " for Task " + taskUri);		
		
		// notify workflow controller of error in task
		notifyTaskError(jobUri, taskUri);		
		
		store.closeStore();
	}
	
	@Override
	public boolean addListener(ActivityControllerListener l) {
		if (l == null)
			return false;
		if (listeners == null)
			listeners = new ArrayList<ActivityControllerListener>();
		if (!listeners.contains(l))
			listeners.add(l);
		return true;
	}

	@Override
	public boolean removeListener(ActivityControllerListener l) {
		if (listeners == null || l == null)
			return false;
		return listeners.remove(l);
	}
	
	@Override
	public void removeAllListeners() {
		if(listeners == null) return;
		listeners.clear();
	}

	@Override
	public Context getContext() {
		return context;
	}
	
}

